package mq

import (
	"errors"
	"github.com/streadway/amqp"
	"sync"
)

type rabbitmq struct {
	conn         *amqp.Connection
	exchangeName string //交换器名称
	exchangeType string //交换器类型 direct-直接模式；fanout-订阅模式;topic-主题模式;x-delayed-message-延迟模式
	key          string //路由键或者队列名称 如果不设置exchange 这个就对应队列名称
	queueName    string //队列名称
	delayTime    int64  //延迟队列延迟时间 单位毫秒
}

func NewRabbitmq(conn *amqp.Connection) *rabbitmq {
	return &rabbitmq{
		conn: conn,
	}
}

// 设置队列名称
func (p *rabbitmq) QueueName(queueName string) *rabbitmq {
	p.queueName = queueName
	return p
}

// 设置交换机名称
func (p *rabbitmq) Exchange(exchangeName string) *rabbitmq {
	p.exchangeName = exchangeName
	return p
}

// 设置交换机名称
func (p *rabbitmq) ExchangeType(exchangeType string) *rabbitmq {
	p.exchangeType = exchangeType
	return p
}

// 设置路由键
func (p *rabbitmq) Key(key string) *rabbitmq {
	p.key = key
	return p
}

// 设置延迟时间
func (p *rabbitmq) DelayTime(delayTime int64) *rabbitmq {
	p.delayTime = delayTime
	return p
}

// 声明队列
func (p rabbitmq) QueueDeclare() error {
	ch, channelErr := p.conn.Channel()
	if channelErr != nil {
		return channelErr
	}
	if p.queueName == "" {
		return errors.New("队列名称不能为空")
	}
	if _, err := ch.QueueDeclare(
		p.queueName, // 队列名称
		true,        // 是否持久化
		false,       // 是否自动删除
		false,       // 是否具有排他性
		false,       // 是否阻塞处理
		nil,         // 其他参数
	); err != nil {
		return err
	}
	return nil
}

// 声明交换机
func (p rabbitmq) QueueExchange(data ...map[string]interface{}) error {
	ch, channelErr := p.conn.Channel()
	if channelErr != nil {
		return channelErr
	}
	if p.exchangeName == "" {
		return errors.New("交换器名称不能为空")
	}
	other := amqp.Table{}
	if len(data) > 0 {
		for k, v := range data[0] {
			other[k] = v
		}
	}
	if p.exchangeType == "" {
		return errors.New("交换器类型不能为空")
	}
	if err := ch.ExchangeDeclare(
		p.exchangeName, // 交换机名称
		p.exchangeType, // 交换机类型
		true,           // 持久化
		false,          // 不自动删除
		false,          // 不内置
		false,          // 不等待服务器响应
		other,          // 额外参数
	); err != nil {
		return err
	}
	return nil
}

// 队列绑定到交换机
func (p rabbitmq) QueueBind() error {
	ch, channelErr := p.conn.Channel()
	if channelErr != nil {
		return channelErr
	}
	if err := ch.QueueBind(
		p.queueName,    // 队列名称
		p.key,          // 路由键为空
		p.exchangeName, // 交换机名称
		false,          // 不等待服务器响应
		nil,            // 额外参数为空
	); err != nil {
		return err
	}
	return nil
}

// 设置channal连接池
func (p rabbitmq) GetPool() *sync.Pool {
	var pool = sync.Pool{
		New: func() interface{} {
			ch, _ := p.conn.Channel()
			return ch
		},
	}
	return &pool
}

// 写入消息
func (p *rabbitmq) Publish(msg string) error {
	ch, channelErr := p.conn.Channel()
	if channelErr != nil {
		return channelErr
	}

	var header = amqp.Table{}
	if p.delayTime > 0 {
		header = amqp.Table{"x-delay": p.delayTime}
	}
	if publishErr := ch.Publish(
		p.exchangeName, // 交换机名称
		p.key,          // 队列名称或者路由键
		false,          // 是否强制
		false,          // 是否立即
		amqp.Publishing{
			Headers:     header,
			ContentType: "text/plain",
			Body:        []byte(msg),
		},
	); publishErr != nil {
		return publishErr
	}
	return nil
}

// 获取队列执行状态
func (p *rabbitmq) GetStats() (amqp.Queue, error) {
	ch, err := p.conn.Channel()
	if err != nil {
		return amqp.Queue{}, err
	}
	defer ch.Close()
	queue, queueDeclarePassiveErr := ch.QueueDeclarePassive(p.key, true, false, false, false, nil)
	if queueDeclarePassiveErr != nil {
		return amqp.Queue{}, queueDeclarePassiveErr
	}
	queueStats, queueInspectErr := ch.QueueInspect(queue.Name)
	if queueInspectErr != nil {
		return amqp.Queue{}, queueInspectErr
	}
	return queueStats, nil
}
